package my.storm.camel;

import my.storm.zmq.ZmqBolt;
import my.storm.zmq.ZmqSpout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;

public class CamelTopology {
	
	private static Logger LOG = LoggerFactory.getLogger(CamelTopology.class);
	
	private static boolean CLUSTER_MODE = Boolean.FALSE;

	public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
		
		int workers = 1;
		if(args.length > 0) {
			if("cluster".equalsIgnoreCase(args[0])) {
				CLUSTER_MODE = Boolean.TRUE;
			} else {
				CLUSTER_MODE = Boolean.FALSE;
			}
			
			if(args.length > 1) {
				try {
					workers = Integer.parseInt(args[1]);
				} catch(Exception e) {
					LOG.warn("second parameters must be numeric");
				}
			}
		}
				
		LOG.info(">>>>> CamelTotpology");
		
		ApplicationContext applicationContext = SpringApplicationFactory.getContext();
		LOG.info("applicationContext={}", applicationContext);
		
		TopologyBuilder builder = new TopologyBuilder();
		
		builder.setSpout("ZmqSpout", new ZmqSpout());
		builder.setBolt("ZmqBolt", new ZmqBolt()).shuffleGrouping("ZmqSpout");
		
		Config conf = new Config();
		conf.setDebug(true);
		
		if(CLUSTER_MODE) {
			conf.setNumWorkers(workers);
			StormSubmitter.submitTopology("storm-camel", conf, builder.createTopology());
			LOG.info(">>>>> topology run in cluster");
		} else {
			conf.setMaxTaskParallelism(workers);
			LocalCluster cluster = new LocalCluster();
			cluster.submitTopology("storm-camel", conf, builder.createTopology());
			LOG.info(">>>>> topology run in local");
		}
	}
}
